from pyspark import SparkContext, SparkConf
from pyspark.mllib.recommendation import MatrixFactorizationModel
import os
import sys

def create_spark_context():
    spark_conf = SparkConf()\
        .setAppName('Pyspark-Recommend-test')\
        .set("spark.driver.extraJavaOptions", "-Xss4096k")
        # .setMaster('local[4]') \

    spark_context = SparkContext(conf=spark_conf) 
    spark_context.setLogLevel('WARN')             
    return spark_context

def prepare_data(spark_context):
    item_rdd = spark_context.textFile("file:/usr/local/ml-100k/u.item")
    # item_rdd = spark_context.textFile("hdfs://hadoop-node1:9000/input/u.item")
    movie_title = item_rdd.map(lambda line: line.split("|")) \
        .map(lambda a: (float(a[0]), a[1]))

    movie_title_dict = movie_title.collectAsMap() 
    return movie_title_dict

def load_model(spark_context):                    
    try:
        # model = MatrixFactorizationModel.load(spark_context, 'hdfs://hadoop-node1:9000/datas/als-model')
        model = MatrixFactorizationModel.load(spark_context, '/usr/local/datas/als-model')
        print (model)
        return model
    except Exception:
            print ("Error Loading")

if __name__ =="__main__": 
    # run->Edit configuration->Script parameters input:--U 198 any user_id
    if len(sys.argv)!=3:
        print("please input parameters:--U user_id,--M movie_id")

def recommend_movies(als, movies, user_id):
    rmd_movies = als.recommendProducts(user_id, 10)
    print('recommend movies:{}'.format(rmd_movies))
    for rmd in rmd_movies:
        print("for user{} recomment movie:{}".format(rmd[0], movies[rmd[1]]))
    return rmd_movies

def recommend_users(als, movies, movie_id):      
    rmd_users = als.recommendUsers(movie_id, 10)
    # print('for ID:{0},movie:{1},user:'.format(movie_id, movies[movie_id]))
    print('for ID:{},movie:{},user:'.format(movie_id, movies[movie_id]))
    for rmd in rmd_users:
        print("ID:{},rating:{}".format(rmd[0], rmd[2]))


def recommend(als_model, movie_dic):
    if sys.argv[1] == '--U':                    
        recommend_movies(als_model, movie_dic, int(sys.argv[2]))
    if sys.argv[1] == '--M':                     
        recommend_users(als_model, movie_dic, int(sys.argv[2]))

if __name__ == "__main__":
    
    # 
    if len(sys.argv) != 3:
        print("please input parameters: --U user_id or --M movie_id")
        exit(-1)
    sc = create_spark_context()

    
    print('============= preparing data =============')
    movie_title_dic = prepare_data(sc)
    print('============= loading model =============')
    als_load_model = load_model(sc)
    print('============= recommend =============')
    recommend(als_load_model, movie_title_dic)
